package com.fulu.game.bigdata.realtime.sink

import java.util.Properties

import com.fulu.game.bigdata.realtime.config.Config
import com.fulu.game.bigdata.realtime.entity.Sink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer
import org.apache.flink.types.Row

object KafkaSinkBuilder {

  def createKafkaSink(sink: Sink) : FlinkKafkaProducer[(Boolean,Row)] = {

    val props = new Properties()
    props.setProperty("bootstrap.servers",Config.getKafkaServer())

    val kafkaProducer = new FlinkKafkaProducer[(Boolean,Row)](sink.getKafkaTopic,new KafkaJsonSerializationSchema(sink),props,FlinkKafkaProducer.Semantic.AT_LEAST_ONCE)

    kafkaProducer
  }
}
